package cn.gxufe.reliable.message.api.consumer;

import cn.gxufe.reliable.message.api.common.RMSApi;
import cn.gxufe.reliable.message.api.mq.ByteRMSConsumer;
import cn.gxufe.reliable.message.common.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.annotation.Transactional;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author yingjie.rong
 **/
public class ProcessMqMessage {

    private static final String sql = "insert into reliable_message(`key`,`header`,`data`,`status`,`id`,`other`) values (?,?,?,?,?,?)";
    private static final ExecutorService executorService =  new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors()/2,
            Runtime.getRuntime().availableProcessors(),
            0L,
            TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>());

    @Autowired
    private ProcessMqMessage processMqMessage;

    public ProcessMqMessage(ApplicationContext applicationContext){
        try {
            applicationContext.getBean(ByteRMSConsumer.class);
        }catch (Throwable e){
            throw new IllegalArgumentException("ByteRMSConsumer in your app must impl !");
        }
    }

    @Autowired
    private MqMessageService handleMessageService;
    @Autowired
    private JdbcTemplate jdbcTemplate;
    @Autowired
    private RMSApi reliableMessageApi;

    private static final  Logger logger = LoggerFactory.getLogger(ProcessMqMessage.class);

    public void handle(Message message){
        Runnable runnable = processMqMessage.handleWithTransactional(message);
        if(runnable != null){
            executorService.submit(() -> {
                try {
                    runnable.run();
                }catch (Exception e){
                    logger.error("send finish or cancel message error ! ",e);
                }
            });
        }
    }

    @Transactional(rollbackFor = Throwable.class)
    public Runnable handleWithTransactional(Message message){
        Message databaseMessage =  getMessage(message);
        // 消费者已经处理成功了，本地不会再调用客户的业务处理流程
        if(databaseMessage != null ){
            if( databaseMessage.getStatus() == Message.STATUS_FINISH) {
                return ()-> reliableMessageApi.finish(databaseMessage);
            } else if ( databaseMessage.getStatus() == Message.STATUS_CANCEL){
                return () -> reliableMessageApi.cancel(databaseMessage);
            } else {
                logger.error("id = {} , key = {} ,status error ! ",databaseMessage.getId(),databaseMessage.getKey(),databaseMessage.getStatus());
            }
            return null;
        } else {
            return process(message);
        }
    }

    private Runnable process(Message message){
        finishOrCancel(message,null);
        Message finish = handleMessageService.handle(message);
        if( finish.getStatus() == Message.STATUS_FINISH ) {
            updateStatus(message.getKey(),Message.STATUS_FINISH);
            return ()-> reliableMessageApi.finish(finish);
        }else if( finish.getStatus() == Message.STATUS_CANCEL){
            updateStatus(message.getKey(),Message.STATUS_CANCEL);
            return ()-> reliableMessageApi.cancel(finish);
        }else {
            throw new IllegalStateException("message status must CANCEL or FINISH for handle from mq ");
        }
    }

    private Message getMessage(Message message){
        Message databaseMessage =  jdbcTemplate.query("select `id`,`key`,`status`,`data`,`header` from reliable_message where `key` = ?",
                new Object[]{Message.TARGET_SERVICE_KEY_PREFIX+message.getKey()} , (ResultSet rs) -> {
                    if (rs.next()) {
                        Message msg = new Message();
                        msg.setId( rs.getLong(1) );
                        msg.setHeader(rs.getString(2));
                        msg.setStatus(rs.getInt(3));
                        msg.setData(rs.getString(4));
                        msg.setKey(rs.getString(5));
                        return msg;
                    }
                    return null;
                });
        return databaseMessage;
    }

    private void finishOrCancel(Message message,String other) {
            List<Object[]> list = new ArrayList<>(1);
            Object[] args = new Object[]{Message.TARGET_SERVICE_KEY_PREFIX + message.getKey(),message.getHeader(),message.getData(),Message.STATUS_CANCEL + Message.STATUS_FINISH ,message.getId(),other};
            list.add(args);
            jdbcTemplate.batchUpdate(sql,list);
    }

    public void updateStatus(String key,int status){
        List<Object[]> list = new ArrayList<>(1);
        Object[] args = new Object[]{status,Message.STATUS_CANCEL + Message.STATUS_FINISH,Message.TARGET_SERVICE_KEY_PREFIX + key};
        list.add(args);
        int[] ints = jdbcTemplate.batchUpdate("update reliable_message set status = ? where status = ? and `key` =  ? ", list);
        if( ints.length == 0 ||  ints[0] != 1){
            throw new IllegalStateException();
        }
    }

}
